package com.yeyouluo.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;
import org.junit.Before;
import org.junit.Test;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;

public class MyHdfsIOClient {
    private FileSystem fs;
    Configuration configuration = new Configuration();

    @Before
    public void init() throws URISyntaxException, IOException, InterruptedException {

        // 1 获取文件系统
        // 配置在集群上运行
        // configuration.set("fs.defaultFS", "hdfs://linux121:9000");
        // FileSystem fs = FileSystem.get(configuration);
        fs = FileSystem.get(new URI("hdfs://linux121:9000"),
                configuration, "hadoop");
    }

    /**
     * 创建目录
     *
     * @throws URISyntaxException
     * @throws IOException
     * @throws InterruptedException
     */
    @Test
    public void testMkdirs() throws IOException {
        fs.mkdirs(new Path("/test"));
    }

    /**
     * 上传文件
     *
     * @throws IOException
     */
    @Test
    public void testPutFileToHDFS() throws IOException {
        // 输入流
        FileInputStream inputStream = new FileInputStream(new File("D:\\lagou.txt"));
        // 输出流
        FSDataOutputStream fsDataOutputStream = fs.create(new Path("/test/lagou.txt"));
//        // 验证写入HDFS是按照packet进行的
//        FSDataOutputStream fsDataOutputStream = fs.create(new Path("/test/lagou.txt"), new Progressable() {
//            @Override
//            public void progress() {
//        // 每个packet(64K)打印一次
//                System.out.println("&");
//            }
//        });
        // 拷贝
        IOUtils.copyBytes(inputStream, fsDataOutputStream, configuration);
        fsDataOutputStream.close();
        inputStream.close();
    }

    /**
     * 下载文件
     *
     * @throws IOException
     */
    @Test
    public void testPullFileFromHDFS() throws IOException {
        // 输入流
        FSDataInputStream fsDataInputStream = fs.open(new Path("/test/lagou.txt"));
        // 输出流
        FileOutputStream outputStream = new FileOutputStream(new File("D:\\lagou-copy.txt"));
        // 下载
        IOUtils.copyBytes(fsDataInputStream, outputStream, configuration);
        outputStream.close();
        fsDataInputStream.close();
    }

    /**
     * seek
     *
     * @throws IOException
     */
    @Test
    public void testSeek() throws IOException {
        // 输入流
        FSDataInputStream fsDataInputStream = fs.open(new Path("/test/lagou.txt"));
        // 下载
        IOUtils.copyBytes(fsDataInputStream, System.out, 4096, false);
        fsDataInputStream.seek(0);
        System.out.println("\n----------------------\n");
        IOUtils.copyBytes(fsDataInputStream, System.out, 4096, false);
        fsDataInputStream.close();
    }


    public void destory() throws IOException {
        // 3 关闭资源
        fs.close();

    }

}
